package com.jiunuan.live.framework.mq.starter.producer;

import com.jiunuan.live.framework.mq.starter.properties.RocketMQProducerProperties;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: JiuNuan
 * @CreateTime: 2024-08-15 15:59
 * @Description: 生产者配置类
 */
@Slf4j
@Configuration
public class RocketMQProducerConfig {
    @Resource
    private RocketMQProducerProperties rocketMQProducerProperties;

    /**
     * 配置并返回一个MQProducer实例
     * 该方法通过@Bean注解指示这是一个Spring Bean，可以被Spring容器管理
     * 通过创建一个自定义的线程池来处理异步发送任务，并配置DefaultMQProducer实例，
     * 设置其连接信息和发送策略，确保消息能够有效地发送到RocketMQ服务器
     *
     * @return 配置好的DefaultMQProducer实例
     */
    @Bean
    public MQProducer mqProducer(){
        // 创建一个线程池用于异步任务执行
        ThreadPoolExecutor asyncThreadPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2, 100, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                // 为线程池中的每个线程设置一个有意义的名字
                return new Thread(r, "rocketmq-async-thread-" + new Random().ints().toString());
            }
        });

        // 创建并配置一个DefaultMQProducer实例
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
        defaultMQProducer.setProducerGroup(rocketMQProducerProperties.getGroupName());
        defaultMQProducer.setNamesrvAddr(rocketMQProducerProperties.getNameSrv());
        defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProducerProperties.getRetryTimes());
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(rocketMQProducerProperties.getRetryTimes());
        defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
        defaultMQProducer.setAsyncSenderExecutor(asyncThreadPool);

        // 尝试启动生产者实例，若失败则抛出运行时异常
        try {
            defaultMQProducer.start();
            // 启动成功后记录日志
            log.info("mq生产者启动成功,nameSrv is {}",rocketMQProducerProperties.getNameSrv());
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }

        // 返回配置好的生产者实例
        return defaultMQProducer;
    }

}
